package com.xzx.mr.tool;

import com.xzx.mr.AnalysisMR;
import com.xzx.mr.io.MysqlOutputFormat;
import com.xzx.mr.kv.AnalysisKey;
import com.xzx.mr.kv.AnalysisValue;
import com.xzx.mr.mapper.MyHBaseMapper;
import com.xzx.mr.reducer.MysqlReducer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.util.Tool;

/**
 * @author xinzhixuan
 * @version 1.0
 * @date 2024/8/10 18:18
 */
public class AnalysisTool implements Tool {
    @Override
    public int run(String[] args) throws Exception {
        Job job = Job.getInstance();
        job.setJarByClass(AnalysisMR.class);
        // mapper
        Scan scan = new Scan();
        scan.setFilter(new SingleColumnValueFilter("d".getBytes(), "isCaller".getBytes(), CompareOperator.EQUAL, new BinaryComparator("1".getBytes())));
        TableMapReduceUtil.initTableMapperJob(
                TableName.valueOf("xzx:call_log"),
                scan,
                MyHBaseMapper.class,
                AnalysisKey.class,
                LongWritable.class,
                job
        );

        //reducer
        job.setReducerClass(MysqlReducer.class);
        job.setOutputKeyClass(AnalysisKey.class);
        job.setOutputValueClass(AnalysisValue.class);

        // output format
        job.setOutputFormatClass(MysqlOutputFormat.class);

        boolean result = job.waitForCompletion(true);
        return result ? 0 : 1;
    }

    @Override
    public void setConf(Configuration conf) {
        conf.set("mapreduce.framework.name", "local");// 必须要设置
        conf.set("mapreduce.app-submission.cross-platform", "true"); // 必须要设置
    }

    @Override
    public Configuration getConf() {
        return null;
    }
}
